#include "I_NetHandler.h"
#include "I_UnixNet.h"
#include "I_UnixNetProcessor.h"
#include "I_UnixNetVConnect.h"

static void
async_signal_hook(EThread *t)
{
	uint64_t counter;
	read(t->evfd, &counter, sizeof(uint64_t));
}

int
NetHandler::startEvent(int event, void *data)
{
	(void)event;

	trigger_event = (Event *)data;
	SET_HANDLER(&NetHandler::mainEvent);
	trigger_event->schedule_every(-ink_hrtime_from_msecond(NET_EVENT_PERIOD));

	return EVENT_CONT;
}

int
NetHandler::mainEvent(int event, void *data)
{
	int poll_timeout = POLL_TIMEOUT_MSECOND;
	EventIO *ep;
	UnixNetVConnect *vc;

	process_enable_list();
	if(!read_ready_list.empty() || !write_ready_list.empty())
	{
		poll_timeout = 0;
	}
	//Polling...
	PollPoller *poller = get_PollPoller(this_ethread());
	poller->pollEvent(poll_timeout);
	PollDescriptor *pd = poller->pd;
	//not have events
	if(pd->result <= 0)
		return EVENT_CONT;

	for(int i = 0; i < pd->result; i++)
	{
		ep = (EventIO *)pd->get_ev_data(i);

		if(ep->type == EVENT_IO_READWRITE)
		{
			if(pd->get_ev_events(i) & (EVENTIO_READ | EVENTIO_ERROR))
			{
				vc = (UnixNetVConnect *)ep->cont;
				vc->read.triggered = true;
				read_ready_list.enqueue(vc);
			}

			if(pd->get_ev_events(i) & (EVENTIO_WRITE | EVENTIO_ERROR))
			{
				vc = (UnixNetVConnect *)ep->cont;
				vc->write.triggered = true;
				write_ready_list.enqueue(vc);
			}
		}else if(ep->type == EVENT_ASYNC_SIGNAL)
		{
			//async signal
			async_signal_hook(trigger_event->ethread);
		}
	}

	process_ready_list();

	return EVENT_CONT;
}

NetHandler::NetHandler()
{
	SET_HANDLER(&NetHandler::startEvent);
}

int
NetHandler::startIO(UnixNetVConnect *vc)
{
	open_connect_list.enqueue(vc);
	int res = 0;

	PollDescriptor *pd = get_PollDescriptor(trigger_event->ethread);
	if(vc->ep.start(pd, vc, EVENTIO_READ|EVENTIO_WRITE) < 0)
	{
		res = errno;
		if(errno != EEXIST)
		{
			//log....faild on EventIO::start
			return -res;
		}
	}

	if(vc->read.triggered)
		read_ready_list.enqueue(vc);

	vc->nh = this;
	return res;
}

void
NetHandler::stopIO(UnixNetVConnect *vc)
{
	open_connect_list.remove(vc);

	vc->ep.stop();

	read_ready_list.remove(vc);
	write_ready_list.remove(vc);

	//because operate enable list must lock this.
	if(vc->read.in_enable_list)
	{
		read_enable_list.remove(vc);
		vc->read.in_enable_list = 0;
	}

	if(vc->write.in_enable_list)
	{
		write_enable_list.remove(vc);
		vc->write.in_enable_list = 0;
	}

	vc->nh = NULL;
}

void
NetHandler::process_enable_list()
{
	UnixNetVConnect *vc = NULL;
	//process read enable list
	while((vc = read_enable_list.dequeue()))
	{
		vc->read.in_enable_list = false;
		if(vc->closed)
		{
			close_UnixNetVConnect(vc);
			continue;
		}
		if(!vc->read.enabled)
			continue;
		vc->ep.modify(EVENTIO_READ);
		if(vc->read.triggered)
			read_ready_list.enqueue(vc);
	}

	//process write enable list
	while((vc = write_enable_list.dequeue()))
	{
		vc->write.in_enable_list = false;
		if(vc->closed)
		{
			close_UnixNetVConnect(vc);
			continue;
		}
		if(!vc->write.enabled)
			continue;
		vc->ep.modify(EVENTIO_WRITE);
		if(vc->write.triggered)
			write_ready_list.enqueue(vc);
	}
}

void
NetHandler::process_ready_list()
{
	UnixNetVConnect *vc;
	while((vc = read_ready_list.dequeue()))
	{
		if(vc->closed)
		{
			close_UnixNetVConnect(vc);
		}else if(vc->read.enabled && vc->read.triggered)
		{
			vc->read_from_net();
		}else if(!vc->read.enabled)
		{
			vc->ep.modify(EVENTIO_READ, false);
		}
	}

	while((vc = write_ready_list.dequeue()))
	{
		if(vc->closed)
		{
			close_UnixNetVConnect(vc);
		}else if(vc->write.enabled && vc->write.triggered)
		{
			vc->write_to_net();
		}else if(!vc->write.enabled)
		{
			vc->ep.modify(EVENTIO_WRITE, false);
		}
	}
}

void close_UnixNetVConnect(UnixNetVConnect *vc)
{
	NetHandler *nh = vc->nh;
	nh->stopIO(vc);
	//close sockfd
	vc->conn.close();
	delete vc;
}
